package lyw.app.dwd.db;

import lyw.app.BaseSqlApp;
import lyw.common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import lyw.util.SQLUtil;


/**
 * @Author lzc
 * @Date 2022/8/19 14:32
 */
public class DwdTradeCartAdd extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeCartAdd().init(
                3004,
                2,
                "DwdTradeCartAdd");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 读取ods_db数据
        readOdsDb(tEnv, "DwdTradeCartAdd");
        
        // 2. 过滤cart_info数据
        Table cartInfo = tEnv.sqlQuery("select " +
                                           "data['id'] id, " +
                                           "data['user_id'] user_id, " +
                                           "data['course_id'] course_id, " +
                                           "data['course_name'] course_name, " +
                                           "data['cart_price'] cart_price, " +
                                           "if(`type` = 'insert', " +
                                           "data['sold'],cast((cast(data['sold'] as int) - cast(`old`['sold'] as int)) as string)) sold, " +
                                           "ts " +
//                                           "pt " +
                                           "from ods_db " +
                                           "where `database`='edu' " +
                                           "and `table`='cart_info' " +
                                           "and (" +
                                           " `type`='insert' " +
                                           "  or (`type`='update'" +
                                           "      and `old`['sold'] is not null " + // 由于sku_num变化导致的update
                                           "      and cast(`data`['sold'] as int) > cast(`old`['sold'] as int)" +
                                           "     )" +
                                           ")");

        
        // 5. 定义一个动态表与kafka的topic关联
        tEnv.executeSql("create table dwd_trade_cart_add( " +
                          "id string, " +
                          "user_id string, " +
                          "course_id string, " +
                          "course_name string, " +
                          "cart_price string, " +
                          "sold string, " +
                          "ts bigint " +
                          ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_CART_ADD)
        );

        cartInfo.executeInsert("dwd_trade_cart_add");
        
        
    }
}
